package cn.com.nes.site.rocketmq.config;

import cn.com.nes.site.rocketmq.strategy.MessageHandlerContext;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.List;

@Component
public class RocketMQConsumerConfig {

    @Value("${rocketmq.nameServer}")
    private String nameServer;

    @Value("${rocketmq.consumer.group}")
    private String consumerGroup;

    @Value("${rocketmq.consumer.topics}")
    private String consumerTopics;

    private DefaultMQPushConsumer consumer;

    // 注入 RocketMQMessageListener 作为监听器
//    private RocketMQListener rocketMQListener;
    private MessageHandlerContext messageHandlerContext;

//    public RocketMQConfig(RocketMQListener rocketMQListener) {
//        this.rocketMQListener = rocketMQListener;
//    }

    public RocketMQConsumerConfig(MessageHandlerContext messageHandlerContext) {
        this.messageHandlerContext = messageHandlerContext;
    }
    @PostConstruct
    public void init() {
        try {
            System.out.println("RocketMQ Config - nameServer: " + nameServer);
            System.out.println("RocketMQ Config - consumerGroup: " + consumerGroup);
            System.out.println("RocketMQ Config - topic: " + consumerTopics);

            consumer = new DefaultMQPushConsumer(consumerGroup);
            consumer.setNamesrvAddr(nameServer);

            // 设置消费的起始位置：从队列的最初位置开始消费
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

            // 订阅多个 Topic
            String[] topicList = consumerTopics.split(",");
            for (String topic : topicList) {
                consumer.subscribe(topic.trim(), "*");
                System.out.println("订阅的 Topic: " + topic);

            }

            // 注册消息监听器，使用策略上下文来处理不同的 topic
            consumer.registerMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext context) {
                    System.out.println("收到消息，消息数量: " + list.size());
                    for (MessageExt message : list) {
                        String topic = message.getTopic();
                        try {
                            // 根据 topic 选择策略
                            messageHandlerContext.handleMessage(topic, message);
                        } catch (Exception e) {
                            e.printStackTrace();
                            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                        }
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });

            // 启动消费者
            consumer.start();
            System.out.println("RocketMQ Consumer Started...");

        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }
}